package com.atguigu.gmall.realtime.util;

import akka.actor.Props;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @author Archie
 * @date 2021-10-17 15:11
 * @description
 */
public class FlinkSourceUtil {

	public static FlinkKafkaConsumer<String> getKafkaSource(String groupId, String topic) {

		Properties props = new Properties();
		props.setProperty("bootstrap.servers", "hadoop101:9092,hadoop102:9092,hadoop103:9092");
		props.setProperty("group.id", groupId);
		props.setProperty("auto.offset.reset", "latest");
		props.setProperty("isolation.level", "read_committed");

		return new FlinkKafkaConsumer<String>(
				topic,
				new SimpleStringSchema(),
				props
		);
	}

}
